package com.icee.myth.common;

import com.icee.myth.common.channelContext.HeartbeatChannelContext;
import com.icee.myth.common.message.serverMessage.Message;
import io.netty.bootstrap.ServerBootstrap;
import io.netty.channel.*;

import java.net.InetSocketAddress;
import java.util.concurrent.LinkedTransferQueue;

/**
 * @author liuxianke
 */
public abstract class AbstractHeartbeatServer implements Server {

    protected String host;    // Server的ip地址
    protected int port; // Server的端口地址
    protected ServerBootstrap bootstrap;
    protected int totalClientNum;
    private Channel bindChannel = null;

    protected EventLoopGroup bossGroup;
    protected EventLoopGroup workGroup;

    protected HeartbeatChannelContext[] channelContexts; // 记录与Client(服务器端相互间的client)连接的channel context，下标是Client的ID，Client的ID从0开始

    protected LinkedTransferQueue<Message> queue;

    public AbstractHeartbeatServer() {
    }

    abstract public void init(String host, int totalClientNum, int port, ChannelHandler childHandler, final LinkedTransferQueue<Message> queue);
//    {
//        this.queue = queue;
//        this.host = host;
//        this.port = port;
//        this.totalClientNum = totalClientNum;
//
//        channelContexts = new HeartbeatChannelContext[totalClientNum];
//        for (int i = 0; i < totalClientNum; i++) {
//            channelContexts[i] = new HeartbeatChannelContext();
//        }
//
//        // Start server with Nb of active threads = 2*NB CPU + 1 as maximum.
//        final int workerCount = Runtime.getRuntime().availableProcessors() * 2 + 1;
//
//        bossGroup = new NioEventLoopGroup();
//        workGroup = new NioEventLoopGroup(workerCount);
//
//        // Configure the server.
//        bootstrap = new ServerBootstrap();
//        bootstrap.group(bossGroup, workGroup);
//        bootstrap.channel(NioServerSocketChannel.class);
//
//        bootstrap.option(ChannelOption.TCP_NODELAY, true);    // 无延迟
//        bootstrap.option(ChannelOption.SO_KEEPALIVE, true);    // 保持存活
//        bootstrap.option(ChannelOption.SO_REUSEADDR, true);    // 重复利用地址
//
//        bootstrap.childHandler(childHandler);
//    }

    @Override
    public void startServer() {
        // Bind and startOnlyOnce to accept incoming connections.
        ChannelFuture future = bootstrap.bind(new InetSocketAddress(host, port));
        future.addListener(new ChannelFutureListener() {
            @Override
            public void operationComplete(ChannelFuture future) throws Exception {
                bindChannel = future.channel();
                System.out.println("bind ok!! port=" + port);
            }
        });
    }

    @Override
    public void closeServer() {
//        bindChannel.close();
        if (bossGroup != null) {
            bossGroup.shutdownGracefully();
        }
        if (workGroup != null) {
            workGroup.shutdownGracefully();
        }
        for (int i = 0; i < totalClientNum; i++) {
            channelContexts[i].close();
        }
    }

    protected void addClient(int clientId, Channel channel) {
        assert (clientId >= 0 && clientId < totalClientNum);
        // TODO: 如果gwId不在范围内应该抛异常

        channelContexts[clientId].setChannel(channel);
        restoreHeartbeat(clientId);
    }

    protected void heartbeat() {
        for (int i = 0; i < totalClientNum; i++) {
            // 向对方发送心跳消息
            channelContexts[i].write(buildServerHeartBeat());

            // 心跳计数减一
            int beatNum = channelContexts[i].heartbeat();
            // 心跳计数小于等于0表示发生故障
            if (beatNum <= 0) {
                // 当心跳计数为0时，向消息队列中产生一服务故障消息。
                if (beatNum == 0) {
//                    ServerMessageQueue.queue().offer(buildClientDownMessage(i));
                    queue.offer(buildClientDownMessage(i));

                    channelContexts[i].close();
                }
                // TODO: 报警，当心跳计数小于等于0后产生连续的报警
                clientDownWarning(i);
            }
        }
    }

    protected void restoreHeartbeat(int clientId) {
        assert (clientId < totalClientNum);
        channelContexts[clientId].restoreHeartbeat();
    }

    protected void clientClose(Channel channel) {
        for (int i = 0; i < totalClientNum; i++) {
            if (channelContexts[i].isChannel(channel)) {
                channelContexts[i].close();
                clientCloseWarning(i);
                return;
            }
        }
    }

    public HeartbeatChannelContext getChannelContext(int id) {
        return channelContexts[id];
    }

    public void broadcast(Object msg) {
        for (int i = 0; i < totalClientNum; i++) {
            channelContexts[i].write(msg);
        }
    }

    public void flush() {
        for (int i = 0; i < totalClientNum; i++) {
            channelContexts[i].flush();
        }
    }

    protected abstract Object buildServerHeartBeat();

    protected abstract Message buildClientDownMessage(int clientId);

    protected abstract void clientDownWarning(int clientId);

    protected abstract void clientCloseWarning(int clientId);
}
